// Copyright 2013 Tumblr, Inc.
// Copyright 2013 Petar Maymounkov
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package hifi

import (
	x "circuit/sys/teleport"
	"encoding/gob"
	"log"
	"math/rand"
	"sync"
)

type linkID int64

func chooseLinkID() linkID {
	return linkID(rand.Int63())
}

type linkOpenMsg struct {
	ID connID
}

type linkConnMsg struct {
	ID      connID
	Payload interface{}
}

func init() {
	gob.Register(&linkOpenMsg{})
	gob.Register(&linkConnMsg{})
}

// link ...
type link struct {
	id       linkID
	addr     x.Addr
	under    ReadWriteCloser
	//under  *permConn
	//dial   *autoDialConn
	//accept *autoAcceptConn
	sync.Mutex
	open map[connID]*conn
}

type diallink struct {
	link
}

type acceptlink struct {
	link
	listener
}

type listener interface {
	OnAccept(*conn)
}

func newDialLink(addr x.Addr) *diallink {
	//a, id := newAutoDialConn(addr)
	l := &diallink{
		link: link{
			id:    chooseLinkID(),
			addr:  addr,
			under: newGobConn(mustDial(addr)),
			open:  make(map[connID]*conn),
			// dial:  a,
			// under: newPermConn(a),
		},
	}
	go l.readloop()
	return l
}

// acceptLink blocks until the initial handshake completes and the identity of the remote is established.
func newAcceptLink(addr x.Addr, id linkID, g *gobConn, listener listener) *acceptlink {
	//a := newAutoAcceptConn(id, g)
	l := &acceptlink{
		link: link{
			id:       id,
			addr:     addr,
			listener: listener,
			under:    g,
			open:     make(map[connID]*conn),
			// accept: a,
			// under:  newPermConn(a),
		},
	}
	go l.readloop()
	return l
}

func (l *link) Dial() *conn {
	id := chooseConnID()
	c := dialConn(l.addr, id, l.under, func() { l.scrub(id) })
	l.attach(id, c)
	return c
}

func (l *link) reserveID(id connID) error {
	l.Lock()
	defer l.Unlock()
	_, present := l.open[id]
	if present {
		return ErrCollision
	}
	l.open[id] = nil // Reserve the slot
	return nil
}

func (l *link) attach(id connID, c *conn) {
	l.Lock()
	defer l.Unlock()
	l.open[id] = c
}

func (l *link) lookup(id connID) *conn {
	l.Lock()
	defer l.Unlock()
	return l.open[id]
}

/*
func (l *link) AcceptRedial(g *gobConn) {
	if l.accept == nil {
		panic("bug")
	}
	l.accept.AcceptRedial(g)
}
*/

func (l *acceptlink) readloop() {
	for {
		m, err := l.under.Read()
		if err != nil {
			return
		}
		switch msg := m.(type) {
		case *linkOpenMsg:
			if err = l.reserveID(msg.ID); err != nil {
				// Connection with colliding ID a protocol violation and
				// indicates a bug in the receiver. We scream about it.
				panic("duplicate open connection IDs")
			}
			c := acceptConn(l.addr, msg.ID, l.under, func() { l.scrub(msg.ID) })
			l.attach(msg.ID, c)
			l.listener.OnAccept(c)

		case *linkConnMsg:
			l.demux(msg)
		default:
			// Drop unknown messages for forward compatibility
			log.Printf("dropping unexpected message type %#T", m)
		}
	}
}

func (l *diallink) readloop() {
	for {
		m, err := l.under.Read()
		if err != nil {
			return
		}
		switch msg := m.(type) {
		case *linkConnMsg:
			l.demux(msg)
		default:
			// Drop unknown messages for forward compatibility
			log.Printf("dropping unexpected message type %#T", m)
		}
	}
}

func (l *link) demux(msg *linkConnMsg) {
	c := l.lookup(msg.ID)
	if c == nil {
		println(msg.ID)
		// Unknown connection. Likely a late packet, arriving after a conn closed on this side. Drop it.
		log.Printf("orphan message")
		return
	}
	if err := c.sendRead(msg.Payload); err != nil {
		l.scrub(c.id)
	}
}

func (l *link) scrub(id connID) {
	l.Lock()
	defer l.Unlock()
	delete(l.open, id)
}

func (l *link) Close() error {
	l.Lock()
	defer l.Unlock()
	return l.under.Close()
}
